package org.leiyang.mq;

import org.leiyang.common.redis.ReactiveRedisCache;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.Subscription;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.Collections;

import static org.leiyang.common.util.Constants.MSG_REC_SUCCESS_GROUP_ID;
import static org.leiyang.common.util.Constants.MSG_REC_SUCCESS_TOPIC;

/**
 * RedisStream配置类
 */
@Configuration
public class RedisStreamConfig {
    private final Logger logger = LoggerFactory.getLogger(getClass());
    @Autowired
    private NewMsgConsumer newMsgConsumer;
    @Autowired
    private OnlineMsgConsumer onlineMsgConsumer;
    @Autowired
    private MsgReceiveConsumer msgReceiveConsumer;
    @Autowired
    private ReactiveRedisCache redisCache;
    @Value("${newMsg.consumer.id}")
    private String newMsgConsumerId;
    @Value("${onlineMsg.consumer.id}")
    private String onlineMsgConsumerId;
    @Value("${successReceiveMsg.consumer.id}")
    private String successReceiveMsgConsumerId;

    @Value("${spring.redis.host}")
    private String redisHost;
    @Value("${spring.redis.port}")
    private Integer redisPort;
    @Value("${spring.redis.password}")
    private String redisPwd;

    private RedisConnectionFactory connectionFactory() {
        RedisStandaloneConfiguration redisStandaloneConfiguration = new RedisStandaloneConfiguration(redisHost, redisPort);
        redisStandaloneConfiguration.setDatabase(0);
        redisStandaloneConfiguration.setPassword(redisPwd);
        LettuceConnectionFactory connectionFactory = new LettuceConnectionFactory(redisStandaloneConfiguration);
        connectionFactory.afterPropertiesSet();
        return connectionFactory;
    }

    @Bean
    public StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String,String,String>> newMsgListenerContainerOptions() {
        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
        return StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                .builder()
                //block读取超时时间
                .pollTimeout(Duration.ofSeconds(3))
                //count数量（一次只获取一条消息）
                .batchSize(1)
                //序列化规则
                .serializer(stringRedisSerializer)
                .build();
    }

    /**
     * 开启{@link NewMsgConsumer#topicName}监听器接收消息
     */
    @Bean
    @Qualifier("newMsgListenerSubscription")
    public Subscription newMsgListenerSubscription(StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String,String,String>> streamMessageListenerContainerOptions){
        StreamMessageListenerContainer<String,MapRecord<String,String,String>> listenerContainer = StreamMessageListenerContainer.create(connectionFactory(),
                streamMessageListenerContainerOptions);
        //如果TOPIC不存在 创建TOPIC
        return redisCache.keyExist(NewMsgConsumer.topicName).flatMap(keyExist -> {
            if(!keyExist) {
                logger.info("初始化stream {} success", NewMsgConsumer.topicName);
                return redisCache.createStreamOperations()
                        .add(NewMsgConsumer.topicName, Collections.singletonMap("field", "value"))
                        .flatMap(recordId -> redisCache.createStreamOperations().delete(NewMsgConsumer.topicName, recordId));
            }
            return Mono.just(1L);
        })
        .flatMap(doResult -> {
            //创建消费者组
            return redisCache.createStreamOperations()
                    .createGroup(NewMsgConsumer.topicName, NewMsgConsumer.groupId);
        })
        .onErrorReturn("failed")
        .map(doResult -> {
            //注册消费者 消费者名称，从哪条消息开始消费，消费者类
            // > 表示没消费过的消息
            // $ 表示最新的消息
            Subscription subscription = listenerContainer.receiveAutoAck(
                    Consumer.from(NewMsgConsumer.groupId, newMsgConsumerId),
                    StreamOffset.create(NewMsgConsumer.topicName, ReadOffset.lastConsumed()),
                    newMsgConsumer
            );
            logger.info("{}的消费者:{}已注册成功.", NewMsgConsumer.topicName, newMsgConsumerId);
            listenerContainer.start();
            return subscription;
        })
        .block();
    }

    /**
     * 开启{@link OnlineMsgConsumer#topicName}的监听器接收消息
     */
    @Bean
    @Qualifier("onlineMsgListenerSubscription")
    public Subscription onlineMsgListenerSubscription(StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String,String,String>> streamMessageListenerContainerOptions){
        StreamMessageListenerContainer<String,MapRecord<String,String,String>> listenerContainer = StreamMessageListenerContainer.create(connectionFactory(),
                streamMessageListenerContainerOptions);
        //如果TOPIC不存在 创建TOPIC
        return redisCache.keyExist(OnlineMsgConsumer.topicName).flatMap(keyExist -> {
            if(!keyExist) {
                logger.info("初始化stream {} success", OnlineMsgConsumer.topicName);
                return redisCache.createStreamOperations()
                        .add(OnlineMsgConsumer.topicName, Collections.singletonMap("field", "value"))
                        .flatMap(recordId -> redisCache.createStreamOperations().delete(OnlineMsgConsumer.topicName, recordId));
            }
            return Mono.just(1L);
        })
        .flatMap(doResult -> {
            //创建消费者组
            return redisCache.createStreamOperations().createGroup(OnlineMsgConsumer.topicName, OnlineMsgConsumer.groupId);
        })
        .onErrorReturn("failed")
        .map(doResult -> {
            //注册消费者 消费者名称，从哪条消息开始消费，消费者类
            // > 表示没消费过的消息
            // $ 表示最新的消息
            Subscription subscription = listenerContainer.receiveAutoAck(
                    Consumer.from(OnlineMsgConsumer.groupId, onlineMsgConsumerId),
                    StreamOffset.create(OnlineMsgConsumer.topicName, ReadOffset.lastConsumed()),
                    onlineMsgConsumer
            );
            logger.info("{}的消费者:{}已注册成功.", OnlineMsgConsumer.topicName, onlineMsgConsumerId);
            listenerContainer.start();
            return subscription;
        })
        .block();
    }

    /**
     * 开启MSG_REC_SUCCESS_TOPIC的监听器接收消息
     */
    @Bean
    @Qualifier("successReceiveMsgListenerSubscription")
    public Subscription successReceiveMsgListenerSubscription(StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String,String,String>> streamMessageListenerContainerOptions){
        StreamMessageListenerContainer<String,MapRecord<String,String,String>> listenerContainer = StreamMessageListenerContainer.create(connectionFactory(),
                streamMessageListenerContainerOptions);
        //如果TOPIC不存在 创建TOPIC
        return redisCache.keyExist(MSG_REC_SUCCESS_TOPIC).flatMap(keyExist -> {
            if(!keyExist) {
                logger.info("初始化stream {} success", MSG_REC_SUCCESS_TOPIC);
                return redisCache.createStreamOperations()
                        .add(MSG_REC_SUCCESS_TOPIC, Collections.singletonMap("field", "value"))
                        .flatMap(recordId -> redisCache.createStreamOperations().delete(MSG_REC_SUCCESS_TOPIC, recordId));

            }
            return Mono.just(1L);
        }).flatMap(doResult -> {
            //创建消费者组
            return redisCache.createStreamOperations().createGroup(MSG_REC_SUCCESS_TOPIC, MSG_REC_SUCCESS_GROUP_ID);
        }).onErrorReturn("failed")
        .map(doResult -> {
            //注册消费者 消费者名称，从哪条消息开始消费，消费者类
            // > 表示没消费过的消息
            // $ 表示最新的消息
            Subscription subscription = listenerContainer.receiveAutoAck(
                    Consumer.from(MSG_REC_SUCCESS_GROUP_ID, successReceiveMsgConsumerId),
                    StreamOffset.create(MSG_REC_SUCCESS_TOPIC, ReadOffset.lastConsumed()),
                    msgReceiveConsumer
            );
            logger.info("{}的消费者:{}已注册成功.", MSG_REC_SUCCESS_TOPIC, successReceiveMsgConsumerId);
            listenerContainer.start();
            return subscription;
        }).block();
    }
}